package com.dreamlin.basepro.exceptions;


import androidx.annotation.NonNull;
import androidx.core.util.Supplier;

import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Completable;
import io.reactivex.CompletableSource;
import io.reactivex.CompletableTransformer;
import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.Maybe;
import io.reactivex.MaybeSource;
import io.reactivex.MaybeTransformer;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.ObservableTransformer;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.reactivex.SingleSource;
import io.reactivex.SingleTransformer;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;


public class GlobalErrorTransform<T> implements ObservableTransformer<T, T>,
        FlowableTransformer<T, T>, SingleTransformer<T, T>,
        MaybeTransformer<T, T>, CompletableTransformer {

    private static Supplier<Scheduler> SCHEDULER_PROVIDER_DEFAULT = AndroidSchedulers::mainThread;

    private Supplier<Scheduler> upStreamSchedulerProvider;
    private Supplier<Scheduler> downStreamSchedulerProvider;
    private Function<T, Observable<T>> globalOnNextRetryInterceptor;
    private Function<Throwable, Observable<T>> globalOnErrorResume;
    private Function<Throwable, RetryConfig> retryConfigProvider;
    private Consumer<Throwable> globalOnErrorConsumer;


    public GlobalErrorTransform(Supplier<Scheduler> upStreamSchedulerProvider,
                                Supplier<Scheduler> downStreamSchedulerProvider,
                                Function<T, Observable<T>> globalOnNextRetryInterceptor,
                                Function<Throwable, Observable<T>> globalOnErrorResume,
                                Function<Throwable, RetryConfig> retryConfigProvider,
                                Consumer<Throwable> globalOnErrorConsumer) {
        this.upStreamSchedulerProvider = upStreamSchedulerProvider;
        this.downStreamSchedulerProvider = downStreamSchedulerProvider;
        this.globalOnNextRetryInterceptor = globalOnNextRetryInterceptor;
        this.globalOnErrorResume = globalOnErrorResume;
        this.retryConfigProvider = retryConfigProvider;
        this.globalOnErrorConsumer = globalOnErrorConsumer;
    }

    public GlobalErrorTransform(Function<T, Observable<T>> globalOnNextRetryInterceptor,
                                Function<Throwable, Observable<T>> globalOnErrorResume,
                                Function<Throwable, RetryConfig> retryConfigProvider,
                                Consumer<Throwable> globalOnErrorConsumer) {
        this(SCHEDULER_PROVIDER_DEFAULT,
                SCHEDULER_PROVIDER_DEFAULT,
                globalOnNextRetryInterceptor,
                globalOnErrorResume,
                retryConfigProvider,
                globalOnErrorConsumer);
    }

    @Override
    public @NonNull
    CompletableSource apply(@NonNull Completable upstream) {
        return upstream
                .observeOn(upStreamSchedulerProvider.get())
                .onErrorResumeNext(throwable -> globalOnErrorResume.apply(throwable).ignoreElements())
                .retryWhen(new FlowRetryDelay(retryConfigProvider))
                .doOnError(globalOnErrorConsumer)
                .observeOn(downStreamSchedulerProvider.get());
    }

    @Override
    public @NonNull
    Publisher<T> apply(@NonNull Flowable<T> upstream) {
        return upstream
                .observeOn(upStreamSchedulerProvider.get())
                .flatMap(throwable -> globalOnNextRetryInterceptor.apply(throwable).toFlowable(BackpressureStrategy.BUFFER))
                .onErrorResumeNext(throwable -> {
                    return globalOnErrorResume.apply(throwable).toFlowable(BackpressureStrategy.BUFFER);
                })
                .retryWhen(new FlowRetryDelay(retryConfigProvider))
                .doOnError(globalOnErrorConsumer)
                .observeOn(downStreamSchedulerProvider.get());
    }

    @Override
    public @NonNull
    MaybeSource<T> apply(@NonNull Maybe<T> upstream) {
        return upstream
                .observeOn(upStreamSchedulerProvider.get())
                .flatMap(throwable -> globalOnNextRetryInterceptor.apply(throwable).firstElement())
                .onErrorResumeNext(throwable -> {
                    return globalOnErrorResume.apply(throwable).firstElement();
                })
                .retryWhen(new FlowRetryDelay(retryConfigProvider))
                .doOnError(globalOnErrorConsumer)
                .observeOn(downStreamSchedulerProvider.get());
    }

    @Override
    public @NonNull
    ObservableSource<T> apply(@NonNull Observable<T> upstream) {
        return upstream
                .observeOn(upStreamSchedulerProvider.get())
                .flatMap(throwable -> globalOnNextRetryInterceptor.apply(throwable))
                .onErrorResumeNext(throwable -> {
                    return globalOnErrorResume.apply(throwable);
                })
                .retryWhen(new ObservableRetryDelay(retryConfigProvider))
                .doOnError(globalOnErrorConsumer)
                .observeOn(downStreamSchedulerProvider.get());
    }

    @Override
    public @NonNull
    SingleSource<T> apply(@NonNull Single<T> upstream) {
        return upstream
                .observeOn(upStreamSchedulerProvider.get())
                .flatMap(throwable -> globalOnNextRetryInterceptor.apply(throwable).firstOrError())
                .onErrorResumeNext(throwable -> globalOnErrorResume.apply(throwable).firstOrError())
                .retryWhen(new FlowRetryDelay(retryConfigProvider))
                .doOnError(globalOnErrorConsumer)
                .observeOn(downStreamSchedulerProvider.get());
    }
}
